package net.xdclass.xdclassmq.jms.producer;

import net.xdclass.xdclassmq.config.JmsConfig;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;
import java.util.concurrent.*;

@Component
public class TransactionProducer {

    @Autowired
    private JmsConfig jmsConfig;

    private static volatile TransactionMQProducer producer = null;

    //事务监听器
    @Autowired
    private TransactionListener transactionListener;

    //一般自定义线程池的时候，需要给线程加个名称
    private ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS,
            new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {
        @Override
        public Thread newThread(Runnable r) {
            //Thread thread = new Thread(); 这种写法导致没有回查
            Thread thread = new Thread(r);
            thread.setName("client-transaction-msg-check-thread");
            return thread;
        }
    });


    public TransactionProducer() {
    }

    public TransactionMQProducer getProducer(){
        if(null == producer){
            synchronized (TransactionProducer.class){
                if(null == producer){
                    producer = new TransactionMQProducer(jmsConfig.getTransProducerGroup());
                    producer.setNamesrvAddr(jmsConfig.getNameServerAddr());
                    producer.setTransactionListener(transactionListener);
                    producer.setExecutorService(executorService);
                    start();
                }
            }
        }
        return producer;
    }


    /**
     * 对象在使用之前必须要调用一次，只能初始化一次
     */
    public void start(){
        try {
            this.producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

    /**
     * 一般在应用上下文，使用上下文监听器，进行关闭
     */
    public @PreDestroy void shutdown(){
        System.out.println("关闭连接");
        this.producer.shutdown();
    }
}
